package com.kingsoft.dc.khaos.module.spark.sink

import java.sql.{Connection, DriverManager, PreparedStatement, Statement}
import java.sql.DriverManager.getConnection
import java.util
import java.util.{Properties, Random}
import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.extender.meta.model.col.DmTableColumn
import com.kingsoft.dc.khaos.extender.meta.model.ds.MysqlConnect
import com.kingsoft.dc.khaos.innertype.Schema
import com.kingsoft.dc.khaos.module.spark.constants.{ColumnType, MetaDataConstants, MysqlConstants, SQLServerConstants, SchedulerConstants}
import com.kingsoft.dc.khaos.module.spark.metadata.sink.{ExtractFieldInfo, MySQLConfig}
import com.kingsoft.dc.khaos.module.spark.model.center.metric.SyncProcessDataMetric
import com.kingsoft.dc.khaos.module.spark.model.{MetaDataEntity, RelationDataStatusInfo}
import com.kingsoft.dc.khaos.module.spark.util._
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DataType, StringType}
import org.apache.spark.sql.{Column, DataFrame, SaveMode}
import org.json4s.DefaultFormats
import org.json4s.JsonAST.JObject
import org.json4s.jackson.JsonMethods.{compact, render}

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

class MysqlSink extends SinkStrategy with Serializable with Logging {

  private var mysqlMeta: MetaDataEntity = null
  private var mysqlConfig: MySQLConfig = null
  private var ip = ""
  private var port = ""
  private var dbName = ""
  private var tblName = ""
  private var dbAndTbl = ""
  private var jdbcUrl = ""
  private var _jdbc_url_param = ""
  private var _jdbc_driver = ""
  private var _has_permission_drop_table: String = ""
  private var executeBatch:Int=10000

  //需要初始化并赋值
  private var ifDeleteOn = false
  private var filter = ""
  private val SINK_MODE_UPSERT = "upsert";

  /** 数据输出 */
  override def sink(kc: KhaosContext,
                    module_id: String,
                    config: JObject,
                    schema: Schema,
                    dataFrame: DataFrame): this.type = {

    var meta_origin: String = kc.conf.getString(SchedulerConstants.META_ORIGIN, MetaDataConstants.META_DATAMANAGEMENT)
    meta_origin match {
      case MetaDataConstants.META_DATAMANAGEMENT =>
        this.write(kc, module_id, config, schema, dataFrame)
      case MetaDataConstants.META_DBSERVER =>
        val dbs: MysqlSinkDbs = new MysqlSinkDbs
        dbs.sink(kc, module_id, config, schema, dataFrame)
      case _ =>
        throw new Exception(s"mysqlSink meta_origin 参数错误! ==> $meta_origin")
    }

    this
  }

  /** 数据输出 */
  def write(kc: KhaosContext,
            module_id: String,
            config: JObject,
            schema: Schema,
            dataFrame: DataFrame): this.type = {
    //log.info(s"mysql writer start ......")
    //log.info(s"mysql writer config json : ${config.toString}")

    init(kc, config)

    val connect = mysqlMeta.dsMysqlConnect
    val colEntiy = mysqlMeta.columnEntiy
    val userName = connect.getUserName
    val passWord = connect.getPassWord
    val writeType = mysqlConfig.write_option.toLowerCase
    val extractFields = mysqlConfig.extract_fields
    ip = connect.getHost
    port = connect.getPort
    dbName = mysqlConfig.db_name
    tblName = mysqlConfig.table_name
    dbAndTbl = s"`$dbName`.`$tblName`"
    ifDeleteOn = mysqlConfig.IfDeleteOn


    var convertDF: DataFrame = DataframeUtils.setDefaultValue(extractFields, colEntiy, dataFrame)
    convertDF = convertDataType(extractFields, convertDF)

    val replaceFieldStrs: String = RmdbUtil.getMysqlCols(colEntiy)
    log.info("获取replaceFieldStrs={}", replaceFieldStrs)

    //将df注册成临时表
    //convertDF.createOrReplaceTempView("MySQLSinkTmp")
    //convertDF.select(selectedFields)

    //对表进行行筛选和列裁剪（预留）
    //val tableQuery = s"select * from MySQLSinkTmp"
    //log.info(s"MySQLSinkTmp sql is: $tableQuery")

    jdbcUrl = s"jdbc:mysql://${ip}:${port}${_jdbc_url_param}"
    val driver = _jdbc_driver
    log.info("MysqlSink driver=" + driver + " url=" + jdbcUrl)
    val prop = new Properties
    prop.put("driver", driver)
    prop.put("user", userName)
    prop.put("password", passWord)
    val osuser = System.getProperty("user.name")
    if (osuser.length > 30)
      prop.put("oracle.jdbc.v$session.osuser", osuser.substring(0, 30))
    //当savemode是overwrite时，truncate在MySQL中的表，而不是删除再重建其现有的表。
    if ("overwrite".equals(writeType)) {
      prop.put("truncate", "true")
    }

    //写入前删除数据按钮是否打开
    if (ifDeleteOn) {
      log.info("ifdeleteon is true")
      filter = mysqlConfig.filter
      val deleteSql: String = deleteTableQuery(dbAndTbl)
      log.info(s"deletesql: ${deleteSql}")
      val desql: Boolean = executeSql(jdbcUrl, userName, passWord, deleteSql)
      if (desql) {
        log.info("删除历史数据成功")
      }
    }


    //获得真实物理表
    val tblNameAndDF: mutable.HashMap[String, DataFrame] = TableSplitUtils.getSinkRealTable(kc, dbName, tblName, this, mysqlConfig.extender.meta.clazz,
      compact(render(mysqlConfig.extender.meta.params)), convertDF, null)
    //    tblNameAndDF = tblNameAndDF.map(tp => (tp._1.toLowerCase(), tp._2))

    //插入的表的字段名称必须和DF中Schema的字段完全一致才能成功,顺序和个数可以不一致
    //分区数默认和传入的DF分区数一致，可再次进行重新分区
    var mysqlDataStatusInfo = new RelationDataStatusInfo
    tblNameAndDF.foreach(tp => {
      val realTblName: String = "`" + tp._1 + "`"
      val realDataframe: DataFrame = tp._2
      val (resultData, accumulator) = DataframeUtils.calculateDataNum(kc, realDataframe, "MysqlSink")
      if (!SINK_MODE_UPSERT.equals(writeType)) {
        resultData.write.mode(writeType).jdbc(jdbcUrl, s"`$dbName`.$realTblName", prop)
      } else {
        //区分两无权限建KTMP_表不同处理逻辑 1：有权限 0:无权限
        if ("1".equals(_has_permission_drop_table)) {
          var adminConn: Connection = null
          var stmt: Statement = null
          //生成临时表"KTMP_" + 时间戳 + 8位随机数
          val tempTable = "KTMP_" + System.currentTimeMillis + "_" + new Random().nextInt(100000000)
          //获取管理员账户信息 只用于建表使用,谨慎使用
          val dsAdmin = MetaUtils.getMysqlDatasourceByName(kc, mysqlConfig.extender.meta.clazz, compact(render(mysqlConfig.extender.meta.params)))
          val mysqlAdminConnect = dsAdmin.getConnect.asInstanceOf[MysqlConnect]
          try {
            Class.forName(driver);
            val adminProp = new Properties
            adminProp.put("driver", driver)
            adminProp.put("user", mysqlAdminConnect.getUserName)
            adminProp.put("password", mysqlAdminConnect.getPassWord)
            adminProp.put("mysql.jdbc.v$session.osuser", osuser.substring(0, 30))
            adminConn = DriverManager.getConnection(jdbcUrl, adminProp)
            stmt = adminConn.createStatement()
            //where 1=0; 这个条件始终为false，结果不会返回任何数据，只有表结构，可用于快速建表
            val sqlCreatTab = s"create table `$dbName`.`$tempTable` as select * from `$dbName`.$realTblName A where 1=0"
            log.info("创建临时表语句===》" + sqlCreatTab)
            //插入原表数据到临时表中
            stmt.execute(sqlCreatTab)
            resultData.write.mode(SaveMode.Append).jdbc(jdbcUrl, s"`$dbName`.`$tempTable`", adminProp)
            resultData.show(10)
            //执行replace into tbl_name(col_name, ...) select ...语句
            val replaceSql =
              s"""
            replace into `$dbName`.`$tblName` ($replaceFieldStrs)
            select $replaceFieldStrs from `$dbName`.`$tempTable`
            """.stripMargin
            log.info(s"============replaceSql:$replaceSql++++++++++++++")
            stmt.execute(replaceSql)
          } finally {
            if (stmt != null) {
              //删除临时表
              stmt.execute(s"drop table if exists `$dbName`.`$tempTable`")
              stmt.close()
              log.info(s"删除临时`$dbName`.`$tempTable` 成功")
            }
            if (adminConn != null)
              adminConn.close()
            log.info(s"关闭$adminConn 连接成功")
          }
        } else {
          realDataframe.rdd.foreachPartition(iter => {
            log.info("线程:{}-------进入foreachPartition",Thread.currentThread().getName)
            var onePartitionJdbcConn: Connection = null
            var oneStmt: Statement = null
            var insertPs: PreparedStatement = null
            //生成临时表"KTMP_" + 时间戳 + 8位随机数
            val tempTable = "KTMP_" + System.currentTimeMillis + "_" + new Random().nextInt(100000000)
            //获取管理员账户信息 只用于建表使用,谨慎使用
            //          val dsAdmin = MetaUtils.getMysqlDatasourceByName(kc, mysqlConfig.extender.meta.clazz, compact(render(mysqlConfig.extender.meta.params)))
            //          val mysqlAdminConnect = dsAdmin.getConnect.asInstanceOf[MysqlConnect]
            Class.forName(driver);
            val adminProp = new Properties
            adminProp.put("driver", driver)
            adminProp.put("user", userName)
            adminProp.put("password", passWord)
            adminProp.put("mysql.jdbc.v$session.osuser", osuser.substring(0, 30))
            onePartitionJdbcConn = DriverManager.getConnection(jdbcUrl, adminProp)
            try {
              onePartitionJdbcConn.setAutoCommit(false) //设置非自动提交

              oneStmt = onePartitionJdbcConn.createStatement()
              //1. where 1=0; 这个条件始终为false，结果不会返回任何数据，只有表结构，可用于快速建临时表
              val sqlCreatTab = s"create TEMPORARY table `$dbName`.`$tempTable` as select * from `$dbName`.$realTblName A where 1=0"
              log.info("创建临时表语句===》" + sqlCreatTab)
              oneStmt.execute(sqlCreatTab)

              //2. 将dataFrame数据转换为行数据插入到临时表中
              val repeat = new Array[Char](colEntiy.size())
              util.Arrays.fill(repeat, '?')
              val repeatStr: String = new String(repeat)
              val selectStr: String = repeatStr.replace("", " ").trim.replaceAll(" ", ",")
              val insertTmpSql = s"insert into `$dbName`.`$tempTable` select " + selectStr
              log.info("创建insert into语句===》" + insertTmpSql)
              insertPs = onePartitionJdbcConn.prepareStatement(insertTmpSql)
              var count = 0
              while (iter.hasNext) {
                val row = iter.next
                val fields = row.schema.fields
                for (i <- 0 until fields.length) {
                  //var dataType: DataType = fields(i).dataType   // println(i + "\t" + fields(i).name + "\t" + fields(i).dataType)
                  val fieldValue = row.get(i)
                  insertPs.setObject(i + 1, fieldValue)
                }
                insertPs.addBatch
                count += 1
                if (0 == count % executeBatch) {
                  log.info("即将提交count量={}", count)
                  insertPs.executeBatch
                }
              } // end while
              insertPs.executeBatch
              //3. 对临时表的数据replace目标mysql表，完成真正的upsert处理
              val replaceSql =
                s"""
            replace into `$dbName`.`$tblName` ($replaceFieldStrs)
            select $replaceFieldStrs from `$dbName`.`$tempTable`
            """.stripMargin
              log.info(s"============replaceSql:$replaceSql++++++++++++++")
              oneStmt.execute(replaceSql)
              onePartitionJdbcConn.commit()
            } catch {
              case e: Exception => {
                log.error("运行upsert过程中报错：", e.getMessage)
                e.printStackTrace()
                onePartitionJdbcConn.rollback()
                throw new Exception(s"mysql upsert writer failed," + e.getMessage)
              }
            } finally {
              if (oneStmt != null) {
                oneStmt.close()
              }
              if (insertPs != null) {
                insertPs.close()
              }
              if (null != onePartitionJdbcConn) {
                onePartitionJdbcConn.close()
                log.info(s"关闭$onePartitionJdbcConn 连接成功")
              }
            }
          }) //end rdd foreach
        }
      } //end mysql upsert
      var numTemp: Long = 0
      if (mysqlDataStatusInfo.getDataNum == null) {
        numTemp = accumulator.value.toLong
      } else {
        numTemp = mysqlDataStatusInfo.getDataNum.toLong + accumulator.value.toLong
      }

      //数据条数累加
      mysqlDataStatusInfo.setDataNum(String.valueOf(numTemp))
    })
    //上报数据状态
    mysqlDataStatusInfo.setCover(if (writeType == "overwrite") true else false)
    DataframeUtils.reportDataStatusRelation(kc, mysqlDataStatusInfo, dbName, tblName, mysqlConfig.extender.meta.clazz,
      compact(render(mysqlConfig.extender.meta.params)))
    //上报运维中心指标
    val metric: SyncProcessDataMetric = CenterMetricUtils.buildSyncProcessDataMetric(kc)
    metric.setProcessDataLValue(mysqlDataStatusInfo.getDataNum.toLong)
    CenterMetricUtils.reportSyncProcessData(metric, kc)
    this
  }

  //生成删除表的sql
  def deleteTableQuery(tblName: String): String = {
    var tableQuery = ""
    if (filter != null && !filter.trim.equals("")) {
      tableQuery = s"delete from $tblName where $filter"
    } else {
      throw new Exception("删除历史数据开关打开后，where条件不能为空！")
    }
    tableQuery
  }

  //执行删除指定数据操作
  def executeSql(url: String, user: String, password: String, sql: String): Boolean = {
    var conn: Connection = null
    val driver = _jdbc_driver
    try {
      Class.forName(driver)
      conn = getConnection(url, user, password)
      val st = conn.createStatement
      for (elem <- sql.split(";", -1).toList) {
        st.addBatch(elem)
      }
      st.executeBatch()
      st.close
      conn.close
      true
    } catch {
      case e: Exception =>
        e.printStackTrace()
        throw e
        false
    }
  }


  /** 初始化参数 */
  def init(kc: KhaosContext, config: JObject): Unit = {
    if (true == false) {
      val str = "21202F2938212B3E22272626252E434D"
    }
    implicit val formats = DefaultFormats
    mysqlConfig = config.extract[MySQLConfig]

    //load config 配置文件参数
    loadProperties(kc)

    //权限校验
    val checkResult = MetaUtils.checkWriteAuth(kc,
      mysqlConfig.db_name,
      mysqlConfig.table_name,
      mysqlConfig.extender.auth.clazz,
      compact(render(mysqlConfig.extender.auth.params)))
    if (!checkResult) {
      log.error(s"mysql writer init failed, 权限验证未通过!")
      throw new Exception(s"mysql writer init failed, 权限验证未通过!")
    }


    //获取元数据
    mysqlMeta = MetaUtils.getMysqlMeta(kc,
      mysqlConfig.db_name,
      mysqlConfig.table_name,
      mysqlConfig.extender.meta.clazz,
      compact(render(mysqlConfig.extender.meta.params)),
      this)


    if (mysqlMeta.tableEntiy.getTblType.equalsIgnoreCase(MetaDataConstants.VIEW)) {
      throw new Exception(s"暂不支持写入视图类型表 ${mysqlConfig.db_name}.${mysqlConfig.table_name}")
    }
  }

  /**
   * load config properties 配置
   *
   * @param kc
   */
  def loadProperties(kc: KhaosContext): Unit = {
    val mysqlProperties: Map[String, String] = kc.conf.getAllWithPrefix("module.mysql.sink.").toMap
    log.info("MysqlSink Properties")
    mysqlProperties.foreach { case (k, v) => log.info(k + "   " + v) }
    _jdbc_url_param = mysqlProperties.getOrElse(MysqlConstants.MODULE_MYSQL_SINK_JDBC_URL_PARAM, "")
    _jdbc_driver = mysqlProperties.getOrElse(MysqlConstants.MODULE_MYSQL_SINK_JDBC_DRIVER, "com.mysql.cj.jdbc.Driver")
    //项目用户是否有权限创建表(用于upsert场景存储临时表数据)
    _has_permission_drop_table = mysqlProperties.getOrElse(MysqlConstants.MODULE_MYSQL_SINK_HASPERMISSION_DROPTABLE, "false")
    executeBatch = Integer.valueOf(mysqlProperties.getOrElse(MysqlConstants.MODULE_MYSQL_SINK_EXECUTEBATCH, "10000"))
  }

  /**
   * 转换DF
   *
   * 修改字段名以及填充默认值
   *
   * @param data
   * @param extractFields
   * @return
   */
  def convertDataFrame(data: DataFrame, extractFields: List[ExtractFieldInfo], columnEntiy: util.List[DmTableColumn]): DataFrame = {
    //log.info(s"MySQLSink before convert DF schema is: ${data.printSchema}")
    //log.info(s"MySQLSink before convert DF is: ${data.show}")
    //生成[字段名,not_null(true/false)]的map
    val fieldAndNotNull = columnEntiy.asScala.map(colEntiy => {
      val colName: String = colEntiy.getColName
      var not_null: String = ""
      colEntiy.getParams.asScala.foreach(map => {
        map.get("pKey") match {
          case "NOT_NULL" => not_null = map.get("pValue")
          case _ =>
        }
      })
      (colName, not_null)
    }).toMap
    //只复制有连线关系的Column
    val colArr = new ArrayBuffer[Column]()
    for (ef <- extractFields) {
      if (!ef.from_field.trim.equals("")) {
        val to_field: String = ef.field
        val from_field: String = ef.from_field
        colArr += data.col(from_field) as (to_field)
      }
    }
    var value: DataFrame = data.select(colArr: _*)
    //对目标DF进行默认值填充和类型转换，填充前需要先转为String类型
    for (ef <- extractFields) {
      val to_field: String = ef.field
      val data_type = ef.data_type
      val field_length = ef.length
      val from_field: String = ef.from_field
      val default_value: String = ef.field_props.default_value
      if (!from_field.trim.equals("")) {
        //将time类型的值去掉空格，否则写入mysql会报错
        if (data_type.equalsIgnoreCase("TIME")) {
          value = value.withColumn(to_field, trim(value.col(to_field).cast(StringType)))
        } else {
          value = value.withColumn(to_field, value.col(to_field).cast(StringType))
        }
      } else {
        value = value.withColumn(to_field, lit(null).cast(StringType))
      }
      //填充默认值
      if (!default_value.equals("")) {
        value = value.na.fill(default_value, Array(to_field))
      } else if (default_value.equals("") && fieldAndNotNull(to_field).equalsIgnoreCase("true") && from_field.trim.equals("")) {
        log.error(s"目标字段：${to_field}不能为null!")
        throw new Exception(s"目标字段：${to_field}不能为null!")
      }
      //将类型转为目标表所需类型
      if (field_length.isEmpty) {
        value = value.withColumn(to_field, value.col(to_field).cast(getDataType(data_type)))
      } else {
        value = value.withColumn(to_field, value.col(to_field).cast(getDataType(data_type, field_length.get)))
      }
    }
    log.info(s"MySQLSink after convert DF schema is: ${value.printSchema}")
    log.info(s"MySQLSink after convert DF is: ${value.show}")
    value
  }

  /** 转换DF字段类型 具体转换查看 getDataType()方法 */
  def convertDataType(sinkSchema: List[ExtractFieldInfo], data: DataFrame): DataFrame = {
    var value: DataFrame = data
    /*for (ef <- sinkSchema) {
      val to_field: String = ef.field
      val data_type = ef.data_type
      val field_length = ef.length
      //将类型转为目标表所需类型
      if (field_length.isEmpty) {
        value = value.withColumn(to_field, value.col(to_field).cast(getDataType(data_type)))
      } else {
        value = value.withColumn(to_field, value.col(to_field).cast(getDataType(data_type, field_length.get)))
      }
    }*/
    val columns: List[Column] = sinkSchema.map((ef: ExtractFieldInfo) => {
      var column: Column = null
      val to_field: String = ef.field
      val data_type: String = ef.data_type
      val field_length: Option[String] = ef.length
      if (field_length.isEmpty) {
        column = value.col(to_field).cast(getDataType(data_type))
      } else {
        column = value.col(to_field).cast(getDataType(data_type, field_length.get))
      }
      column
    })

    value.select(columns: _*)
  }

  /** 获取目标字段中对应的值的类型，无字段长度参数 */
  def getDataType(dataType: String): String = {
    var value: String = null
    // `string`, `boolean`, `byte`, `short`, `int`, `long`, * `float`, `double`, `decimal`, `date`, `timestamp`.
    value = dataType match {
      case ColumnType.DATE => "date"
      case ColumnType.TIME => "string"
      case ColumnType.DATETIME => "timestamp"
      case _ => "string"
    }
    value
  }

  /** 获取目标字段中对应的值的类型，有字段长度参数 */
  def getDataType(dataType: String, fieldLength: String): String = {
    var value: String = null
    // `string`, `boolean`, `byte`, `short`, `int`, `long`, * `float`, `double`, `decimal`, `date`, `timestamp`.
    value = dataType match {
      case ColumnType.NUMBER => "long"
      case ColumnType.STRING => "string"
      case ColumnType.DECIMAL => s"decimal($fieldLength)"
      case ColumnType.DATE => "date"
      case ColumnType.TIME => "string"
      case ColumnType.DATETIME => "timestamp"
      case _ => "string"
    }
    value
  }
}